package com.mq.demo.listener;

import com.alibaba.fastjson.JSONObject;
import lombok.extern.java.Log;
import org.apache.rocketmq.spring.annotation.RocketMQTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionState;
import org.springframework.messaging.Message;

/**
 * @Author LY
 * @create 2023/7/18 16:43
 * @Description rockermq 事物消息
 */
@RocketMQTransactionListener(rocketMQTemplateBeanName = "rocketMQTemplate")
@Log
public class RocketTransactionListener implements RocketMQLocalTransactionListener {

    @Override
    public RocketMQLocalTransactionState executeLocalTransaction(Message msg, Object arg) {
        //执行本地事务，在第一步中消息发送成功后回调执行，一旦事务提交成功，下游应用的Consumer能收到该消息，在这里demo的本地事务就是保持订单信息入库
        log.info("开始执行本地事务 executeLocalTransaction ===>>>");
        String messageString = new String((byte[]) msg.getPayload());
        JSONObject jsonObject = JSONObject.parseObject(messageString);
        log.info("消息内容：msg=" + jsonObject + ",arg=" + arg);
        Integer itr = (Integer) arg;
        if (itr == 1){
            return RocketMQLocalTransactionState.COMMIT;
        }else if (itr == 2){
            return RocketMQLocalTransactionState.UNKNOWN;
        }
        return RocketMQLocalTransactionState.ROLLBACK;
    }

    @Override
    public RocketMQLocalTransactionState checkLocalTransaction(Message msg) {
        // 检查本地事务执行状态，如果executeLocalTransaction方法中返回的状态是未知或者未返回状态，默认会在预处理发送的1分钟后由Broker通知Producer检查本地事务，
        // 在Producer中回调本地事务监听器中的checkLocalTransaction方法。检查本地事务时，可以根据事务ID查询本地事务状态，返回具体事务状态给Broker
        log.info("开始回查本地事务状态 checkLocalTransaction ===>>>");
        String messageString = new String((byte[]) msg.getPayload());
        JSONObject jsonObject = JSONObject.parseObject(messageString);
        log.info("消息内容：msg=" + jsonObject);
        return RocketMQLocalTransactionState.COMMIT;
    }

}
